package net.wlm.realtime.new_job;

import com.alibaba.fastjson.JSON;
import net.wlm.realtime.bean.VehicleData;
import net.wlm.realtime.utils.DateTimeUtil;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * 异常检测与告警
 */

public class RealtimeVehicleAEDS {
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 数据源
        DataStream<String> vehicleData = KafkaUtil.consumerKafka(env, "vehicle-data");
        // 数据清洗
        DataStream<VehicleData> vehicleStream = cleanData(vehicleData);
        // vehicleStream.print();
        // 数据转换
        DataStream<String> stringDataStream = handle(vehicleStream);
        stringDataStream.print();

        env.execute();
    }

    private static DataStream<String> handle(DataStream<VehicleData> vehicleStream) {
        // 最大速度
        final double MaxSpeed = 60.0;
        // 最大电池温度
        final double MaxBatterTemp = 30.0;
        // 最大电机温度
        final double MaxMotorTemp = 60.0;
        // 最低电池电量
        final double MinBatterLevel = 10.0;
        // 设置水位线
        SingleOutputStreamOperator<VehicleData> watermarks = vehicleStream.assignTimestampsAndWatermarks(WatermarkStrategy.<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(new SerializableTimestampAssigner<VehicleData>() {
                    @Override
                    public long extractTimestamp(VehicleData vehicleData, long l) {
                        return vehicleData.getTimestamp();
                    }
                }));
        // 分组
        KeyedStream<VehicleData, String> vehicleDataStringKeyedStream = watermarks.keyBy(a ->a.getVin());
        // CEP,连续三次温度过高
        Pattern<VehicleData, VehicleData> pattern = Pattern.<VehicleData>begin("o").where(
                new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData event) {
                        return event.getBatteryTemp() > MaxBatterTemp;
                    }
                }
        ).next("t").subtype(VehicleData.class).where(
                new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData subEvent) {
                        return subEvent.getBatteryTemp() > MaxBatterTemp;
                    }
                }
        ).next("th").where(
                new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData subEvent) {
                        return subEvent.getBatteryTemp() > MaxBatterTemp;
                    }
                }
        ).within(Time.minutes(5));
        // 执行匹配
        PatternStream<VehicleData> patternStream = CEP.pattern(vehicleDataStringKeyedStream, pattern);
        // 获取结果
        DataStream<String> result = patternStream.process(
                new PatternProcessFunction<VehicleData, String>() {
                    @Override
                    public void processMatch(Map<String, List<VehicleData>> map, Context ctx, Collector<String> out) throws Exception {
                        VehicleData oneData = map.get("o").get(0);
                        VehicleData twoData = map.get("t").get(0);
                        VehicleData threeData = map.get("th").get(0);

                        // 时间
                        String startTime = DateTimeUtil.convertLongToString(oneData.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);
                        String endTime = DateTimeUtil.convertLongToString(threeData.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);


                        // 返回报警信息
                        String alarmData = startTime+ endTime +twoData.getVin()+" Warning";
                        out.collect(alarmData);
                    }
                });
        // 其余报警
        SingleOutputStreamOperator<String> process = vehicleDataStringKeyedStream.process(new ProcessFunction<VehicleData, String>() {
            @Override
            public void processElement(VehicleData vehicleData, Context ctx, Collector<String> out) throws Exception {
                String vin = vehicleData.getVin();
                String timestamp = DateTimeUtil.convertLongToString(vehicleData.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);
                Double speed = vehicleData.getSpeed();
                Double batteryLevel = vehicleData.getBatteryLevel();
                Double motorTemp = vehicleData.getMotorTemp();
                if (speed > MaxSpeed) {
                    out.collect(timestamp + timestamp  + vin + "已超速");
                }
                if (batteryLevel > MinBatterLevel) {
                    out.collect(timestamp + timestamp  + vin + "电量过低");
                }
                if (motorTemp > MaxMotorTemp) {
                    out.collect(timestamp + timestamp  + vin + "电机温度过高");
                }
            }
        });
        return result.union(process);
    }

    private static DataStream<VehicleData> cleanData(DataStream<String> vehicleData) {
        DataStream<VehicleData> dataStream = vehicleData.filter(json -> {
                    try {
                        return json != null && JSON.isValid(json);
                    } catch (Exception e) {
                        return false;
                    }
                })
                // 将合法JSON字符串转换为VehicleData对象
                .map(json -> JSON.parseObject(json, VehicleData.class));
        return dataStream;
    }
}
